package com.atguigu.gmall.util;

import com.atguigu.gmall.bean.TransientSink;
import lombok.SneakyThrows;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @author : ranzlupup
 * @since : 2023/6/4 20:56
 */
public class MyClickHouseUtils {
    public static <T> SinkFunction<T> getSinkFunction(String sql) {
        return JdbcSink.<T>sink(
                sql,
                new JdbcStatementBuilder<T>() {
                    @SneakyThrows
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                        // 使用反射方式获取t对象中的数据
                        Class<?> tClass = t.getClass();

                        // 获取且遍历属性
                        Field[] fields = tClass.getDeclaredFields();
                        int offset = 0;
                        for (int i = 0; i < fields.length; i++) {
                            // 获取单个属性
                            Field field = fields[i];
                            field.setAccessible(true);

                            //尝试获取字段上的自定义注解
                            TransientSink transientSink = field.getAnnotation(TransientSink.class);
                            if (transientSink != null) {
                                offset++;
                                continue;
                            }

                            // 获取属性值
                            Object value = field.get(t);

                            // 给占位符赋值
                            preparedStatement.setObject(i + 1 - offset, value);
                        }

                    }
                },
                new JdbcExecutionOptions.Builder()
                        .withBatchSize(5)
                        .withBatchIntervalMs(1000L)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(MyConfig.CLICKHOUSE_DRIVER)
                        .withUrl(MyConfig.CLICKHOUSE_URL)
                        .build());
    }
}
